/*
package com.zlx.rulemk.marketing.functions;

import com.zlx.rulemk.marketing.beans.*;
import com.zlx.rulemk.marketing.utils.FlinkKeyUtils;
import com.zlx.rulemk.marketing.utils.FlinkStateDescUtils;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;


*/
/**
 * @Author: zhanglingxing
 * @CreateTime: 2022-07-10
 * @Description: 一条日志可能对应多条规则，进行数据扩展
 * @Version: 1.0
 *  这个类的目的就是动态分组
 *      一条日志可能对应多条规则，进行数据扩展
 *//*

@Slf4j
public class DynamicKeyByReplicationFunction extends BroadcastProcessFunction<EventBean, CDCLogBean, DynamicKeyedBean> {

    private Integer flink_parallelism;
    public DynamicKeyByReplicationFunction(Integer flink_parallelism){
        flink_parallelism = this.flink_parallelism;
    }

    */
/*
      动态keyby数据复制
     *//*

    @Override
    public void processElement(EventBean eventBean, ReadOnlyContext ctx, Collector<DynamicKeyedBean> out) throws Exception {

        ReadOnlyBroadcastState<String, List<DynamicKeyRuleStateBean>> broadcastState = ctx.getBroadcastState(FlinkStateDescUtils.ruleStateDesc1);

        if(broadcastState == null) return;

        for (Map.Entry<String,  List<DynamicKeyRuleStateBean>> ruleStateEntry : broadcastState.immutableEntries()) {

            // 一个事件 可能配置多条规则  多条不同的分组情况
            List<DynamicKeyRuleStateBean> dynamicKeyRuleStateBeanList = ruleStateEntry.getValue();

            for (DynamicKeyRuleStateBean dynamicKeyRuleStateBean : dynamicKeyRuleStateBeanList ) {

                String keyByFields = dynamicKeyRuleStateBean.getKeyByFields();

                // 规则要求的触发事件
                String ruleTriggerEventId = dynamicKeyRuleStateBean.getRuleTriggerEventId();

                String ruleName = dynamicKeyRuleStateBean.getRuleName();

                // 只有规则触发事件等于触发事件的时候 我们才会动态分组
                if(ruleTriggerEventId.equals(eventBean.getEventId())){

                    //获取分组key对应的具体value的值
                    StringBuilder sb = new StringBuilder();
                    String[] fieldNames = keyByFields.split(",");
                    // 拼装keyByFields中指定的每一个字段的值
                    for (String fieldName : fieldNames) {
                        Class<?> beanClass = Class.forName("com.zlx.rulemk.marketing.beans.EventBean");
                        Field declaredField = beanClass.getDeclaredField(fieldName);
                        declaredField.setAccessible(true);
                        // 从eventbean中取分组字段的值
                        String fieldValue = (String)declaredField.get(eventBean);
                        sb.append(fieldValue).append(",");
                    }
                    // 获取具体的分组字段的
                    String keyByValue = sb.substring(0, sb.length() - 1);
                    eventBean.setKeyByValue(keyByValue);

                    DynamicKeyedBean dynamicKeyedBean = new DynamicKeyedBean(keyByValue, keyByFields, eventBean,ruleName, FlinkKeyUtils.getKey(flink_parallelism,ruleName));

                    // 完成了 动态数据扩展 一条日志 可能对应多条规则 进行了数据扩张
                    out.collect(dynamicKeyedBean);
                }
            }
        }
    }

    */
/**
     * 插入mysql规则数据进行触发下面的方法
     *//*

    @Override
    public void processBroadcastElement(CDCLogBean CDCLogBean, Context ctx, Collector<DynamicKeyedBean> out) throws Exception {

        BroadcastState<String, List<DynamicKeyRuleStateBean>> broadcastState = ctx.getBroadcastState(FlinkStateDescUtils.ruleStateDesc1);

        // 规则平台上操作的规则的数据
        List<RuleTableRecord> ruleTableRecords = CDCLogBean.getData();
        if(ruleTableRecords == null ) return;

        // 规则平台上的操作的类型
        String operationType = CDCLogBean.getType();

        // 根据不同的操作类型，去操作state中的规则数据
        for (RuleTableRecord ruleTableRecord : ruleTableRecords) {

            String rule_condition_json = ruleTableRecord.getRule_condition_json();
            MarketingRule marketingRule = JSON.parseObject(rule_condition_json, MarketingRule.class);
            String keyByFields = marketingRule.getKeyByFields();
            // 事件ID 我们的规则都是针对某一个事件配置的规则
            String triggerEventId = marketingRule.getTriggerEventCondition().getEventId();
            // 这条规则的唯一ID
            String rule_name = ruleTableRecord.getRule_name();

            List<DynamicKeyRuleStateBean> dynamicKeyRuleStateBeans = broadcastState.get(triggerEventId);
            if(dynamicKeyRuleStateBeans == null){
                dynamicKeyRuleStateBeans = new ArrayList<>();
            }
            // 新增规则
            if ("INSERT".equals(operationType)  ) {
                DynamicKeyRuleStateBean dynamicKeyRuleStateBean = new DynamicKeyRuleStateBean(triggerEventId,rule_name,keyByFields);
                dynamicKeyRuleStateBeans.add(dynamicKeyRuleStateBean);
                broadcastState.put(triggerEventId,dynamicKeyRuleStateBeans);
            }
            // 更新规则
            if("UPDATE".equals(operationType) && "1".equals(ruleTableRecord.getRule_status())){
                for (DynamicKeyRuleStateBean stateBean : dynamicKeyRuleStateBeans) {
                    if(stateBean.getRuleName().equals(rule_name)){
                        stateBean.setKeyByFields(keyByFields);
                        stateBean.setRuleTriggerEventId(triggerEventId);
                        break;
                    }
                }
            }
            if ("DELETE".equals(operationType) || ("UPDATE".equals(operationType) && "0".equals(ruleTableRecord.getRule_status()))) {
                // 如果该keybyfields对应的ruleList已经被删空，则直接从广播state中将这个keybyFields条目都删除
                if(dynamicKeyRuleStateBeans.size() == 0){
                    return ;
                }else if(dynamicKeyRuleStateBeans.size() == 1){
                    //移除整个规则
                    broadcastState.remove(triggerEventId);
                }else{
                    // 删除具体的一条规则
                    Iterator<DynamicKeyRuleStateBean> iterator = dynamicKeyRuleStateBeans.iterator();
                    while (iterator.hasNext()){
                        DynamicKeyRuleStateBean stateBean = iterator.next();
                        if(stateBean.getRuleName().equals(rule_name)){
                            //删除该规则
                            iterator.remove();
                            break;//跳出while循环
                        }
                    }
                }
            }
            log.debug("操作规则，规则对应的triggerEventId：{},简化规则:{}",triggerEventId,dynamicKeyRuleStateBeans);
        }

    }

}
*/
